package com.example.demo.four;

import com.example.demo.util.RabbitMQUtils;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布确认模式
 * 1. 单个确认
 * 2. 批量确认
 * 3. 异步批量确认
 *
 * @author chensixiang chensixiang1234@gmail.com
 */
public class ConfirmMessage {

    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        // publish1(); //703ms
        // publish2(); // 138ms
        publish3(); // 36ms
    }

    public static void publish1() throws IOException, InterruptedException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        long start = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 单个消息马上进行发布确认
            boolean b = channel.waitForConfirms();
            if (b) {
                System.out.println("消息发送成功");
            }
        }

        System.out.println("发布" + MESSAGE_COUNT
                + "条消息" + "耗时" + (System.currentTimeMillis() - start) + "ms");

    }

    public static void publish2() throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        long start = System.currentTimeMillis();

        int batchSize = 100;

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 单个消息马上进行发布确认

            if (i % batchSize == 0) {
                boolean b = channel.waitForConfirms();
                if (b) {
                    System.out.println("消息发送成功");
                }
            }
        }


        System.out.println("发布" + MESSAGE_COUNT
                + "条消息" + "耗时" + (System.currentTimeMillis() - start) + "ms");
    }

    // 异步发布确认
    public static void publish3() throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        /**
         * 线程安全的跳表适合用于高并发的情况下
         * 1. 序号与消息关联
         * 2. 轻松的批量删除
         * 3. 支持高并发
         */
        ConcurrentSkipListMap<Long, Object> map = new ConcurrentSkipListMap<>();


        // 消息监听器
        channel.addConfirmListener(
                (deliveryTag, multiple) -> {
                    // 成功
                    if(multiple) {
                        ConcurrentNavigableMap<Long, Object> res =
                                map.headMap(deliveryTag);
                        res.clear();
                    } else {
                        map.remove(deliveryTag);
                    }

                    System.out.println("确认的消息" + deliveryTag);
                }, (deliveryTag, multiple) -> {
                    //失败
                    String o = (String) map.get(deliveryTag);
                    System.out.println("content" + o);
                    System.out.println("未确认的消息" + deliveryTag);
                });

        long start = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            map.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", queueName, null, message.getBytes());
        }
        System.out.println("发布" + MESSAGE_COUNT
                + "条消息" + "耗时" + (System.currentTimeMillis() - start) + "ms");
    }
}
